package com.pccc.aiopm.csup.redis;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.JedisCluster;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

public class BatchToRedis {

    /**
     * 说明：应该lib是不需要这么多jar包的，但是为了方便都加上去了
     * @param args
     * @throws IOException
     * @throws InterruptedException
     * @throws ClassNotFoundException
     */


    public static void main(String[] args) throws IOException, InterruptedException, ClassNotFoundException {

        Configuration configuration = new Configuration();
        FileSystem fileSystem = FileSystem.get(configuration);

        Job job = Job.getInstance(configuration);
        job.setJarByClass(BatchToRedis.class);
        job.setJobName("batch to redis job");
        job.setMapperClass(BatchToRedisMapper.class);
        FileInputFormat.addInputPath(job,new Path("/input"));

        TextInputFormat.setMinInputSplitSize(job,268435456L);

        Path outputPath = new Path("/output");

        if (fileSystem.exists(outputPath)) {
            fileSystem.delete(outputPath, true);
        }

        FileOutputFormat.setOutputPath(job, outputPath);

        if (job.waitForCompletion(true)){
            System.out.println(job.getJobName() + "执行成功！");
        }

    }

    public static class BatchToRedisMapper extends Mapper<LongWritable, Text, Text, Text> {

        public static JedisCluster jedisCluster = null;

        static {
            // 创建JedisCluster节点集合
            Set<HostAndPort> jedisClusterNodes = new HashSet<>();
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7000));
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7001));
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7002));
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7003));
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7004));
            jedisClusterNodes.add(new HostAndPort("192.168.195.202", 7005));

            // 创建JedisCluster实例
            jedisCluster = new JedisCluster(jedisClusterNodes);
        }


        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context) throws IOException, InterruptedException {
            super.map(key, value, context);

            String[] cardIndexSplit = value.toString().replace("\"","").split(",", -1);

            // 存储redis信息
            // 从数据中获取APPUID和CUSTID，分别是第16个和第2个
            String pan = cardIndexSplit[0];
            String csr_rfe_nbr = cardIndexSplit[2];

            String scl_scy_typ_cde = "0102";
            String scl_scy_nbr = cardIndexSplit[4];
//
            String mob_phe = cardIndexSplit[5];
            String app_uid = cardIndexSplit[16];

//            jedisCluster.set(pan, csr_rfe_nbr);
//            jedisCluster.set(scl_scy_typ_cde + "_" + scl_scy_nbr, csr_rfe_nbr);
//            jedisCluster.set(mob_phe, csr_rfe_nbr);
//            jedisCluster.set(app_uid, csr_rfe_nbr);
//
//            jedisCluster.hset(csr_rfe_nbr, pan, "6222525115213020,281510119203111,000190011057055,SSNO,331102201510286122,13003690467,281552985215015,1050141130024333562,20220015893347,1050141137452753005,281552985666663,1050141130024777773,281552375354323,1050141130014999993,281552985111113,1050107042707474203,1005073760");

            // 关闭连接（这个要在哪里写）
//            try {
//                jedisCluster.close();
//            } catch (IOException e) {
//                throw new RuntimeException(e);
//            }
        }



    }




}